package pub.tbc.rubbish.client.registry.impl;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import pub.tbc.rubbish.client.registry.Registry;
import pub.tbc.toolkit.Objs;

import java.util.concurrent.TimeUnit;

/**
 * 练习zk->使用zookeeper(zk官方提供的JAVA-API)进行注册
 *
 * @author tbc on 2016/12/8 10:37:54.
 */
public class ZkApiRegistry implements Registry {
    @Override
    public void register(String serviceName, String serviceAddress) {

        ZooKeeper client = ZKAPIHelper.getClient();
        //ZooKeeper客户端和服务端会话的建立是异步的；
        // 构造方法会在处理完客户端初始化工作后立即返回，此时，并没有真正建立一个可用的会话，在会话生命周期中处于"CONNECTING"状态；
        // 会话真正创建后，会向会话对应的客户端发一个事件通知，告知会话建立成功

        while (true) {
            System.out.println("=> " + client.getState().isConnected());
            if (client.getState().isConnected()) {
                // 判断并创建节点
                try {
                    Stat stat = client.exists(getRegistryNode(), new Watcher() {
                        @Override
                        public void process(WatchedEvent event) {
                            System.out.println(event.getPath() + " : " + event.getType());
                            System.out.println(event.getPath() + " : " + event.getState());

                        }
                    });
                    if (Objs.isEmpty(stat)) {
                        // 节点不存在，创建
                        String node = getRegistryNode();

                        client.create(
                                "/" + node.split("/")[1],
                                "".getBytes(),
                                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                                CreateMode.PERSISTENT
                        );

                        client.create(
                                node,
                                "".getBytes(),
                                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                                CreateMode.PERSISTENT
                        );

                        // 创建临时节点
                        client.create(
                                getRegistryNode() + "/" + serviceName,
                                "".getBytes(),
                                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                                CreateMode.EPHEMERAL
                        );
                        client.create(
                                getRegistryNode() + "/" + serviceName + "/" + serviceAddress,
                                "".getBytes(),
                                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                                CreateMode.EPHEMERAL
                        );
                    }
                } catch (KeeperException | InterruptedException e) {
                    throw new RuntimeException(e.getMessage(), e);
                }
            }
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private String getRegistryNode() {
        return "/rubbish/registry";
    }


    static class ZKAPIHelper {
        static ZooKeeper getClient() {
            ZooKeeper client = null;
            try {
                client = new ZooKeeper(
                        "127.0.0.1:2181",
                        5000,
                        new Watcher() {
                            @Override
                            public void process(WatchedEvent watchedEvent) {
                                System.out.println("path    -> " + watchedEvent.getPath());
                                System.out.println("state   -> " + watchedEvent.getState());
                                System.out.println("type    -> " + watchedEvent.getType());
                                System.out.println("wrapper -> " + watchedEvent.getWrapper());
                            }
                        }
                );
                System.out.println("over new client, and state : " + client.getState());
            } catch (Exception e) {
                throw new RuntimeException(e.getMessage(), e);
            }
            return client;
        }
    }


    public static void main(String[] args) throws InterruptedException {
        Registry registry = new ZkApiRegistry();
        registry.register("rt", "192.168.1.186:8080");
        TimeUnit.SECONDS.sleep(10);
    }
}

